package com.my.code.pool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 手动拆分任务
 */
public class ExecutorAcc implements Accumulation {

    private int parallism;
    private ExecutorService pool;

    public ExecutorAcc() {
        this.parallism = Runtime.getRuntime().availableProcessors();
        this.pool = Executors.newFixedThreadPool(parallism);
        System.out.println("ExecutorAcc init parallism = " + parallism);
    }

    private static class Task implements Callable<Long> {

        private long[] numbs;
        private int from;
        private int to;

        public Task(long[] numbs, int from, int to) {
            this.numbs = numbs;
            this.from = from;
            this.to = to;
        }

        @Override
        public Long call() throws Exception {
            long total = 0;
            for (int i = from; i <= to; i++) {
                total += numbs[i];
            }
            return total;
        }
    }

    @Override
    public long acc(long[] numbs) {
        List<Future<Long>> results = new ArrayList<>();

        //手动拆分任务
        int part = numbs.length / parallism;
        for (int i = 0; i < parallism; i++) {
            int from = i * part;
            int to = (i == parallism - 1) ? numbs.length - 1 : (i + 1) * part - 1;
            results.add(pool.submit(new Task(numbs, from, to)));
        }

        long total = 0;
        for (Future<Long> future : results) {
            try {
                total += future.get();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return total;
    }
}
